package drds.global_transaction.api.rpc;

import com.alibaba.fastjson.JSON;
import drds.common.Constants;
import drds.common.Log;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.RequestMessages;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.resource_manager.RegisterResourceManagerResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_begin.GlobalTransactionBeginRequestMessage;
import drds.global_transaction.api.rpc.config.ClientConfig;
import drds.global_transaction.api.rpc.rpc_message.RpcInfo;
import drds.global_transaction.api.rpc.rpc_message.RpcInfoFuture;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.thread.NamedThreadFactory;
import io.netty.channel.*;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public abstract class AbstractChannelDuplexHandler extends ChannelDuplexHandler
{
	private static final long not_writeable_check_mills = 10L;
	private static final int timeout_check_internal = 3000;
	//
	protected final ScheduledExecutorService scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("timeoutChecker", 1, true));
	protected final ThreadPoolExecutor threadPoolExecutor;
	//
	private final Object lock = new Object();
	@Setter
	@Getter
	public ConcurrentHashMap<Long, RpcInfoFuture> rpcInfoFutureIdToRpcInfoFutureMap = new ConcurrentHashMap<Long, RpcInfoFuture>();
	//
	@Setter
	@Getter
	public ConcurrentHashMap<String, BlockingQueue<RpcInfo>> addressToRpcInfoQueueMap = new ConcurrentHashMap<String, BlockingQueue<RpcInfo>>();
	@Setter
	@Getter
	public Object addressToRpcInfoQueueMapLock = new Object();
	@Setter
	@Getter
	public boolean isSending = false;
	@Setter
	@Getter
	public Map<Long, RequestMessages> requestMessagesIdToRequestMessagesMap = new ConcurrentHashMap<Long, RequestMessages>();
	//

	@Setter
	@Getter
	public volatile long nowDateTimeLong = 0;

	@Setter
	@Getter
	public ChannelHandler[] channelHandlers;
	@Setter
	@Getter
	private String group = "default";

	public AbstractChannelDuplexHandler(ThreadPoolExecutor threadPoolExecutor)
	{
		this.threadPoolExecutor = threadPoolExecutor;
	}

	protected void addChannelHandlersToChannelPipelineLast(Channel channel, ChannelHandler... channelHandlers)
	{
		if (null != channel && null != channelHandlers)
		{
			channel.pipeline().addLast(channelHandlers);
		}
	}

	public void init()
	{
		scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				List<RpcInfoFuture> timeoutRpcInfoFutureList = new ArrayList<RpcInfoFuture>(rpcInfoFutureIdToRpcInfoFutureMap.size());
				for (RpcInfoFuture rpcInfoFuture : rpcInfoFutureIdToRpcInfoFutureMap.values())
				{
					if (rpcInfoFuture.isTimeout())
					{
						timeoutRpcInfoFutureList.add(rpcInfoFuture);
					}
				}

				for (RpcInfoFuture rpcInfoFuture : timeoutRpcInfoFutureList)
				{
					rpcInfoFutureIdToRpcInfoFutureMap.remove(rpcInfoFuture.getRpcInfo().getId());
					rpcInfoFuture.setTimeOutMark(true);
					rpcInfoFuture.setResponseMessage(null);
					//
					log.warn("timeout clear future : " + rpcInfoFuture.getRpcInfo().getMessage());
				}
				nowDateTimeLong = System.currentTimeMillis();
			}
		}, timeout_check_internal, timeout_check_internal, TimeUnit.MILLISECONDS);
	}

	//
	// 单向请求和单向响应
	public void synchronousSendRequest(Channel channel, Object message)
	{
		RpcInfo rpcInfo = new RpcInfo();
		rpcInfo.setSynchronous(true);
		rpcInfo.setHeartbeat(message instanceof HeartbeatMessage);
		rpcInfo.setRequest(true);// request
		rpcInfo.setMessage(message);
		rpcInfo.setId(RpcInfo.getNextMessageId());
		if (message instanceof RequestMessages)
		{
			requestMessagesIdToRequestMessagesMap.put(rpcInfo.getId(), (RequestMessages) message);
		}
		channelWriteableCheck(channel);
		channel.writeAndFlush(rpcInfo);
	}

	protected void synchronousSendResponse(Channel channel, long messageId, Object message)
	{
		RpcInfo rpcInfo = new RpcInfo();
		rpcInfo.setSynchronous(true);
		rpcInfo.setHeartbeat(message instanceof HeartbeatMessage);
		rpcInfo.setRequest(false);
		rpcInfo.setMessage(message);
		rpcInfo.setId(messageId);
		channelWriteableCheck(channel);
		channel.writeAndFlush(rpcInfo);
	}

	private void channelWriteableCheck(Channel channel)
	{
		int tryTimes = 0;
		synchronized (lock)
		{
			while (!channel.isWritable())
			{
				try
				{
					tryTimes++;
					if (tryTimes > ClientConfig.getMaxNotWriteableRetry())
					{
						returnChannel(channel);
						throw new FrameworkException("channel is not writable", FrameworkErrorCode.ChannelIsNotWritable);
					}
					lock.wait(not_writeable_check_mills);
				} catch (InterruptedException exx)
				{
					log.error(exx.getMessage());
				}
			}
		}
	}

	//
	public Object asynchronousSendRequestWithDefaultTimeoutGetResponse(String address, Channel channel, Object message) throws TimeoutException
	{
		return asynchronousSendRequestWithTimeoutGetResponse(address, channel, message, ClientConfig.getRpcRequestTimeout());
	}

	protected Object asynchronousSendRequestWithTimeoutGetResponse(String address, Channel channel, Object message, long timeout) throws TimeoutException
	{
		if (timeout <= 0)
		{
			throw new FrameworkException("timeout should more than 0ms");
		}
		return asynchronousSendRequestWithCanTimeoutGetResponse(address, channel, message, timeout);
	}

	protected Object asynchronousSendRequestWithNoTimeoutGetResponse(String address, Channel channel, Object message) throws TimeoutException
	{
		return asynchronousSendRequestWithCanTimeoutGetResponse(address, channel, message, 0);
	}

	private Object asynchronousSendRequestWithCanTimeoutGetResponse(String address, @NonNull Channel channel, Object message, long timeout) throws TimeoutException
	{
		if (Constants.developMode)
		{
			if (message instanceof GlobalTransactionBeginRequestMessage)
			{
				String msg=GlobalTransactionBeginRequestMessage.class.getSimpleName() + "来了"+channel.localAddress().toString();
				log.info(msg);
				System.out.println(msg);
			}
		}
		//这里是最底层发送信息的地方...
		final RpcInfo rpcInfo = new RpcInfo();
		rpcInfo.setId(RpcInfo.getNextMessageId());
		rpcInfo.setSynchronous(false);
		rpcInfo.setHeartbeat(false);
		rpcInfo.setRequest(true);
		rpcInfo.setMessage(message);

		final RpcInfoFuture rpcInfoFuture = new RpcInfoFuture();
		rpcInfoFuture.setRpcInfo(rpcInfo);
		rpcInfoFuture.setTimeout(timeout);
		//
		rpcInfoFutureIdToRpcInfoFutureMap.put(rpcInfo.getId(), rpcInfoFuture);
		// why need address,this is the reason 'rpcInfoQueue.offer(rpcInfo);'
		if (address != null)
		{
			BlockingQueue<RpcInfo> rpcInfoQueue = addressToRpcInfoQueueMap.get(address);
			if (rpcInfoQueue == null)
			{
				addressToRpcInfoQueueMap.putIfAbsent(address, new LinkedBlockingQueue<RpcInfo>());
				rpcInfoQueue = addressToRpcInfoQueueMap.get(address);
			}
			rpcInfoQueue.offer(rpcInfo);
			// notice to 'send message thread' become running
			if (!isSending)
			{
				// notice deal with these message list
				synchronized (addressToRpcInfoQueueMapLock)
				{
					addressToRpcInfoQueueMapLock.notifyAll();
				}
			}
		} else
		{
			channelWriteableCheck(channel);
			ChannelFuture channelFuture = channel.writeAndFlush(rpcInfo);// right now write
			channelFuture.addListener(new ChannelFutureListener()
			{
				@Override
				public void operationComplete(ChannelFuture channelFuture)
				{
					if (!channelFuture.isSuccess())
					{
						RpcInfoFuture rpcInfoFuture = rpcInfoFutureIdToRpcInfoFutureMap.remove(rpcInfo.getId());
						if (rpcInfoFuture != null)
						{
							rpcInfoFuture.setResponseMessage(channelFuture.cause());
						}
						returnChannel(channelFuture.channel());
					}
				}
			});
		}
		if (timeout > 0)
		{
			try
			{
				Object object = rpcInfoFuture.get(timeout, TimeUnit.MILLISECONDS);
				if (Constants.developMode)
				{
					log.debug("服务器返回信息:" + object);
					if (object != null)
					{
						log.debug(JSON.toJSONString(object));
						//
						if (object instanceof RegisterTransactionManagerResponseMessage)
						{
							RegisterTransactionManagerResponseMessage registerTransactionManagerResponseMessage = (RegisterTransactionManagerResponseMessage) object;
							Log.debugWithLine(log, Thread.currentThread(), "注册事务管理器返回信息:" + registerTransactionManagerResponseMessage.isSuccess());
						}
						if (object instanceof RegisterResourceManagerResponseMessage)
						{
							RegisterResourceManagerResponseMessage registerResourceManagerResponseMessage = (RegisterResourceManagerResponseMessage) object;
							Log.debugWithLine(log, Thread.currentThread(), "注册资源管理器返回信息:" + registerResourceManagerResponseMessage.isSuccess());
						}
						//
					}
				}
				return object;
			} catch (Exception e)
			{
				log.error("wait response error:" + e.getMessage() + ",ip:" + address + ",request:" + message);
				if (e instanceof TimeoutException)
				{
					throw (TimeoutException) e;
				} else
				{
					throw new RuntimeException(e);
				}
			}
		} else
		{
			return null;
		}
	}

	//
	@Override
	public void channelRead(final ChannelHandlerContext channelHandlerContext, Object message) throws Exception
	{
		if (message instanceof RpcInfo)
		{
			final RpcInfo rpcInfo = (RpcInfo) message;
			if (rpcInfo.isRequest())
			{
				// request
				if (Constants.developMode)
				{
					log.debug(String.format("%s msgId:%s, body:%s", this, rpcInfo.getId(), rpcInfo.getMessage()));
				}
				//
				try
				{
					AbstractChannelDuplexHandler.this.threadPoolExecutor.execute(new Runnable()
					{
						@Override
						public void run()
						{
							try
							{
								dispatch(channelHandlerContext, rpcInfo.getId(), rpcInfo.getMessage());
							} catch (Exception e)
							{
								log.error(FrameworkErrorCode.NetDispatch.errCode, e.getMessage(), e);
							}
						}
					});
				} catch (RejectedExecutionException e)
				{
					log.error(FrameworkErrorCode.ThreadPoolFull.errCode, "thread pool is full, current max pool size is " + threadPoolExecutor.getActiveCount());
				}
			} else
			{
				// response
				RpcInfoFuture rpcInfoFuture = rpcInfoFutureIdToRpcInfoFutureMap.remove(rpcInfo.getId());
				if (Constants.developMode)
				{
					log.debug(String.format("%s msgId:%s, future :%s, body:%s", this, rpcInfo.getId(), rpcInfoFuture, rpcInfo.getMessage()));
				}
				if (rpcInfoFuture != null)
				{
					rpcInfoFuture.setResponseMessage(rpcInfo.getMessage());
				} else
				{
					try
					{
						AbstractChannelDuplexHandler.this.threadPoolExecutor.execute(new Runnable()
						{
							@Override
							public void run()
							{
								try
								{
									dispatch(channelHandlerContext, rpcInfo.getId(), rpcInfo.getMessage());
								} catch (Exception e)
								{
									log.error(FrameworkErrorCode.NetDispatch.errCode, e.getMessage(), e);
								}
							}
						});
					} catch (RejectedExecutionException e)
					{
						log.error(FrameworkErrorCode.ThreadPoolFull.errCode, "thread pool is full, current max pool size is " + threadPoolExecutor.getActiveCount());
					}
				}
			}
		}
	}

	public abstract void dispatch(ChannelHandlerContext channelHandlerContext, long messageId, Object message);

	//
	@Override
	public void channelWritabilityChanged(ChannelHandlerContext channelHandlerContext)
	{
		synchronized (lock)
		{
			if (channelHandlerContext.channel().isWritable())
			{
				lock.notifyAll();
			}
		}

		channelHandlerContext.fireChannelWritabilityChanged();
	}

	@Override
	public void close(ChannelHandlerContext channelHandlerContext, ChannelPromise channelPromise) throws Exception
	{
		super.close(channelHandlerContext, channelPromise);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) throws Exception
	{
		try
		{
			returnChannel(channelHandlerContext.channel());
		} catch (Exception e)
		{
			log.error("", "close channel" + channelHandlerContext.channel() + " fail.", e);
		}
		log.error(FrameworkErrorCode.ExceptionCaught.errCode, channelHandlerContext.channel() + " connect exception. " + cause.getMessage(), cause);
	}

	public void destroy()
	{
		scheduledThreadPoolExecutor.shutdown();
	}

	//

	protected String getRemoteAddress(ChannelHandlerContext channelHandlerContext)
	{
		return getRemoteAddress(channelHandlerContext.channel());
	}

	protected String getRemoteAddress(Channel channel)
	{
		SocketAddress socketAddress = channel.remoteAddress();
		String address = socketAddress.toString();
		if (socketAddress.toString().indexOf(ClientConfig.getSocketAddressStartChar()) == 0)
		{
			address = socketAddress.toString().substring(ClientConfig.getSocketAddressStartChar().length());
		}
		return address;
	}
	//

	//
	public void returnChannel(Channel channel)
	{
		returnChannel(getRemoteAddress(channel), channel);
	}

	public abstract void returnChannel(String serverAddress, Channel channel);

}
